package com.apps.sdses.flink141.udfs;

import com.apps.sdses.flink141.entity.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

/**
 * @author jiwei
 * @description
 * @date 2023/6/5 15:04
 */
public class Flink_UDF_ScalarFunction {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        env.setParallelism(3);
        DataStreamSource<WaterSensor> waterSensorStream =
                env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                        new WaterSensor("sensor_1", 2000L, 20),
                        new WaterSensor("sensor_2", 3000L, 30),
                        new WaterSensor("sensor_1", 4000L, 40),
                        new WaterSensor("sensor_1", 5000L, 50),
                        new WaterSensor("sensor_2", 6000L, 60));

        Table inputTable = tEnv.fromDataStream(waterSensorStream);

        tEnv.createTemporarySystemFunction("SubStringFunction", SubstringFunction.class);


        Table resultTable = tEnv.sqlQuery("select SubStringFunction(id, 7, 8), SubStringFunction(vc) from " + inputTable + " where id = 'sensor_1'");

        tEnv.toDataStream(resultTable, Row.class).print();

        env.execute();
    }


    // 定义函数逻辑
    public static class SubstringFunction extends ScalarFunction {
        public String eval(String s, Integer begin, Integer end) {
            return s.substring(begin, end);
        }

        public Integer eval(Integer vc) {
            return vc + 1;
        }

    }
}
